package com.bluemoon.model.test.disrupter;

import cn.com.bluemoon.daps.model.core.file.transformer.ImportEntryTableDto;
import com.google.common.collect.Lists;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.util.SerializationUtils;

import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

/**
 * @author Jarod.Kong
 * @date 2021/3/18 9:27
 */
public class DisrupterClient {

    public static void main(String[] args) {
        // 预数据
        List<ModelTableEvent> modelTableEvents = Lists.newArrayList();
        IntStream.range(0, 1000000)
        .forEach(i ->{
            ModelTableEvent event = new ModelTableEvent();
            event.setEntryModel(ImportEntryTableDto.builder().build());
            modelTableEvents.add(event);
        });
        // 1.创建一个可缓存的线程 提供线程来出发Consumer 的事件处理
        ExecutorService executor = Executors.newCachedThreadPool();
        // 2.创建工厂
        EventFactory<ModelTableEvent> eventFactory = new ModelTableEventFactory();
        // 3.创建ringBuffer 大小
        int ringBufferSize = 1024 * 1024; // ringBufferSize大小一定要是2的N次方
        // 4.创建Disruptor
        Disruptor<ModelTableEvent> disruptor = new Disruptor<ModelTableEvent>(
                eventFactory, ringBufferSize, executor,
                ProducerType.SINGLE, new YieldingWaitStrategy());
        // 5.连接消费端方法
        disruptor.handleEventsWith(new ModelTableEventHandler());
        // 6.启动
        disruptor.start();
        // 7.创建RingBuffer容器
        RingBuffer<ModelTableEvent> ringBuffer = disruptor.getRingBuffer();
        // 8.创建生产者
        ModelTableEventProducer producer = new ModelTableEventProducer(ringBuffer);
        // 9.指定缓冲区大小
        ByteBuffer buffer = ByteBuffer.allocate(1024);

        for (ModelTableEvent modelTableEvent : modelTableEvents) {
            byte[] serialize = SerializationUtils.serialize(modelTableEvent);
            assert serialize != null;
            buffer.put(serialize);
            producer.onData(buffer);
            buffer.clear();

        }
        //10.关闭disruptor和executor
        disruptor.shutdown();
        executor.shutdown();
    }
}


